package signal;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.webrtc.SessionDescription;

import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;

import json_object_mapper.Trickle;
import json_object_mapper.TrickleJsonRPC;
import json_object_mapper.request_rpc_objects.AnswerJsonRPC;
import json_object_mapper.request_rpc_objects.Description;
import json_object_mapper.request_rpc_objects.JoinJsonRPC;
import json_object_mapper.request_rpc_objects.JoinRequestParam;
import json_object_mapper.request_rpc_objects.OfferJsonRPC;
import json_object_mapper.request_rpc_objects.SessionDescriptionSerializer;
import json_object_mapper.response_rpc_objects.SDPAnswerResponseRPC;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okio.ByteString;


/**
 * @author namvdo
 */

public class JSonRPCImpl implements Signal {
    private static final Logger logger = LoggerFactory.getLogger(JSonRPCImpl.class);
    private final ObjectMapper objectMapper = new ObjectMapper();
    private WebSocket webSocket;
    private Consumer<Trickle> onTrickle;
    private Consumer<SessionDescription> onNegotiation;

    private SDPAnswerResponseRPC answerForJoin;
    private SDPAnswerResponseRPC answerForPublish;
    private boolean isJoinAnswerReceived = false;

    enum SignalMessage {
        JOIN("join"),
        OFFER("offer"),
        ANSWER("answer"),
        TRICKLE("trickle"),
        METHOD("method"),
        RESULT("result"),
        ERROR("error"),
        CODE("code"),
        MESSAGE("message"),
        PARAMS("params"),
        ID("id");
        String msg;

        SignalMessage(String msg) {
            this.msg = msg;
        }
    }

    public JSonRPCImpl(String url) {
        startWebSocketClient(url);
    }

    @Override
    public Future<SessionDescription> join(String sid, String uid, SessionDescription offer) {
        String uuid = UUID.randomUUID().toString();
        SessionDescriptionSerializer sessionDescriptionSerializer = new SessionDescriptionSerializer(offer.type, offer.description);
        JoinRequestParam joinReqParams = new JoinRequestParam(sid, uid, sessionDescriptionSerializer);
        JoinJsonRPC joinJsonRPC = new JoinJsonRPC(SignalMessage.JOIN.msg, uuid, joinReqParams);
        try {
            String joinReq = objectMapper.writeValueAsString(joinJsonRPC);
            logger.info("Join request: {}", joinReq);
            webSocket.send(joinReq);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return CompletableFuture.completedFuture(answerForJoin.getResult().getAsSessionDescription());
    }

    @Override
    public Future<SessionDescription> offer(SessionDescription offer) {
        String uuid = UUID.randomUUID().toString();
        OfferJsonRPC offerJsonRpc = new OfferJsonRPC();
        offerJsonRpc.setId(uuid);
        offerJsonRpc.setMethod(SignalMessage.OFFER.msg);
        SessionDescriptionSerializer sessionDescriptionSerializer = new SessionDescriptionSerializer(offer.type, offer.description);
        Description description = new Description(sessionDescriptionSerializer);
        offerJsonRpc.setParams(description);
        try {
            String offerReq = objectMapper.writeValueAsString(offerJsonRpc);
            webSocket.send(offerReq);
            logger.info("Send offer to the moon: {}", offer.description);
            Thread.sleep(2000);
        } catch (JsonProcessingException | InterruptedException e) {
            e.printStackTrace();
        }
        return CompletableFuture.completedFuture(answerForPublish.getResult().getAsSessionDescription());
    }

    @Override
    public void answer(SessionDescription answer) {
        AnswerJsonRPC answerJsonRpc = new AnswerJsonRPC();
        answerJsonRpc.setMethod(SignalMessage.ANSWER.msg);
        SessionDescriptionSerializer sessionDescriptionSerializer = new SessionDescriptionSerializer(answer.type, answer.description);
        Description description = new Description(sessionDescriptionSerializer);
        answerJsonRpc.setParams(description);
        try {
            String answerReq = objectMapper.writeValueAsString(answerJsonRpc);
            logger.info("Answer sent to server {}", answerReq);
            webSocket.send(answerReq);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void trickle(Trickle trickle) {
        String trickleReq;
        TrickleJsonRPC trickleMsg = new TrickleJsonRPC();
        trickleMsg.setParams(trickle);
        try {
            trickleReq = objectMapper.writeValueAsString(trickleMsg);
            if (trickleReq != null) {
                webSocket.send(trickleReq);
            }
            logger.info("trickle sent: {}", trickleReq);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() {
        closeWebSocket();
    }

    public void onTrickle(Trickle trickle) {
        if (this.onTrickle != null) {
            logger.info("Consume onTrickle");
            this.onTrickle.accept(trickle);
        }
    }

    @Override
    public void setOnTrickle(Consumer<Trickle> onTrickle) {
        this.onTrickle = onTrickle;
    }

    public void onNegotiate(SessionDescription sdp) {
        if (this.onNegotiation != null) {
            this.onNegotiation.accept(sdp);
        }
    }

    @Override
    public void setOnNegotiate(Consumer<SessionDescription> onNegotiation) {
        this.onNegotiation = onNegotiation;
    }


    private final class SocketListener extends okhttp3.WebSocketListener {


        @Override
        public void onOpen(okhttp3.WebSocket webSocket, Response response) {
            logger.info("Successfully connect to web socket.");
        }

        @Override
        public void onMessage(okhttp3.WebSocket webSocket, String message) {
            JsonObject resp = JsonParser.parseString(message).getAsJsonObject();
            logger.info("Message received from server: {}", message);
            if (resp.get(SignalMessage.METHOD.msg) != null) {
                String method = resp.get(SignalMessage.METHOD.msg).getAsString();
                if (SignalMessage.TRICKLE.msg.equals(method)) {
                    Trickle trickle = getTrickle(message);
                    onTrickle(trickle);
                } else if (SignalMessage.OFFER.msg.equals(method)) {
                    SessionDescription sdp = getSessionDescription(message);
                    onNegotiate(sdp);
                }

            } else if (resp.get(SignalMessage.ERROR.msg) != null) {
                    String code = resp.get(SignalMessage.ERROR.msg).getAsJsonObject().get(SignalMessage.CODE.msg).getAsString();
                    String errorMsg = resp.get(SignalMessage.ERROR.msg).getAsJsonObject().get(SignalMessage.MESSAGE.msg).getAsString();
                    logger.error("Error code: {}, msg{}", code, errorMsg);

            } else if (resp.get(SignalMessage.ID.msg) != null || resp.get(SignalMessage.RESULT.msg) != null) {
                try {
                    SDPAnswerResponseRPC answer = objectMapper.readValue(message, SDPAnswerResponseRPC.class);
                    if (!isJoinAnswerReceived) {
                        answerForJoin = answer;
                        isJoinAnswerReceived = true;
                    } else {
                        answerForPublish = answer;
                    }
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
            }

        }

        private Trickle getTrickle(String responseMsg) {
            JsonObject resp = JsonParser.parseString(responseMsg).getAsJsonObject();
            Trickle trickle = null;
            String trickleIce = resp.get(SignalMessage.PARAMS.msg).getAsJsonObject().toString();
            try {
                trickle = objectMapper.readValue(trickleIce, Trickle.class);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            return trickle;
        }

        private SessionDescription getSessionDescription(String responseMsg) {
            JsonObject resp = JsonParser.parseString(responseMsg).getAsJsonObject();
            SessionDescription sdp = null;
            SessionDescription.Type type = SessionDescription.Type
                    .valueOf(resp.get("params").getAsJsonObject().get("type").getAsString().toUpperCase());
            String sdpStr = resp.get("params").getAsJsonObject().get("sdp").getAsString();
            sdp = new SessionDescription(type, sdpStr);
            return sdp;
        }

        @Override
        public void onMessage(okhttp3.WebSocket webSocket, ByteString bytes) {

        }

        @Override
        public void onClosing(okhttp3.WebSocket webSocket, int code, String reason) {
            final int NORMAL_CLOSURE_STATUS = 1000;
            webSocket.close(NORMAL_CLOSURE_STATUS, null);
            logger.error("Web socket closed!");
        }

        @Override
        public void onFailure(okhttp3.WebSocket webSocket, Throwable t, Response response) {
            logger.error("Fail to connect to web socket: {}", response);
        }


    }


    private void startWebSocketClient(String url) {
        Request request = new Request.Builder().url(url).build();
        OkHttpClient client = new OkHttpClient();
        SocketListener socketListener = new SocketListener();
        webSocket = client.newWebSocket(request, socketListener);
    }

    private void closeWebSocket() {
        webSocket.close(1000, "");
    }


}



